/**
 * 2017年7月10日
 */
package cn.edu.bjtu.model.core;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.datavec.api.berkeley.Pair;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.textplantform.common.Constant;
import org.textplantform.common.api.AfterInitializationExecute;
import org.textplantform.common.api.LabeledRecord;
import org.textplantform.common.api.RecordWriter;
import org.textplantform.common.api.core.Result;
import org.textplantform.common.api.mq.MessageConsumer;
import org.textplantform.common.api.mq.PlantformTextConsumer;
import org.textplantform.common.core.ClassificationPair;
import org.textplantform.common.core.LoggerSupport;
import org.textplantform.common.core.mq.BasePlantformKafkaConsumer;
import org.textplantform.common.core.mq.KafkaConsumerFactory;
import org.textplantform.common.provider.ClassificationService;

import cn.edu.bjtu.api.ScheduledClassificationService;


/**
 * @author Alex
 *
 */
public class ModelMessageQueueHelper extends LoggerSupport implements AfterInitializationExecute,DisposableBean,BeanFactoryAware {
	ClassificationService cs;
	//这个只是当初始化失败的时候推迟重试.在自己有一个独立的线程池感觉浪费,所以就新加入了一个接口.
	ScheduledExecutorService ss;
	RecordWriter recordWriter;
	
	protected PlantformTextConsumer<ConsumerRecord<String,String>> consumer;
	private class MQConsumer implements MessageConsumer<ConsumerRecord<String,String>>{
		private final ClassificationService cs;
		List<Pair<String, String>> batch = new ArrayList<Pair<String,String>>(100);
		List<LabeledRecord> result = new ArrayList<LabeledRecord>(100);
		public MQConsumer() {
			this.cs = ModelMessageQueueHelper.this.cs;
		}
		@Override
		public void consume(Iterable<ConsumerRecord<String, String>> messages) {
			for(ConsumerRecord<String, String> cr:messages){
				batch.add(Pair.makePair(cr.key(), cr.value()));
			}
			try {
				//1. 将batch数据分类
				String docs[] = new String[batch.size()];
				for(int i=0;i<docs.length;i++){
					docs[i] = batch.get(i).getSecond();
				}
				ClassificationPair [][] res = cs.classifyDocument(docs);
				for(int i=0;i<res.length;i++){
					result.add(new R(batch.get(i).getFirst(),batch.get(i).getSecond(),Arrays.toString(res[i])));
				}
				result.forEach(x->logger.info(x.toString()));
				//2. 将分类好之后的数据存储到mongodb
				Result r = recordWriter.writeLabeledRecord(result.iterator());
				if(r.hasException()){
					logException(r.getException());
				}
			} catch (Exception e) {
				logException(e);
			}finally{
				batch.clear();
				result.clear();
			}
		}
		@Override
		public void consume(ConsumerRecord<String, String> message) {
			throw new RuntimeException(" not implemented ");
		}
	}
	public ModelMessageQueueHelper(ScheduledClassificationService cs){
		ss = cs.getScheduledES();
		this.cs = cs;
		
	}
	@Override
	public void executeAfterBeanInitialization() {
		initMQ();
	}
	private void initMQ(){
		try{
			this.recordWriter = bf.getBean(RecordWriter.class);
			consumer = KafkaConsumerFactory.getConsumer(Constant.Values.MQ_TOPIC);
		}catch(Exception e){
			logger.info(" failed to connect to kafka ....  {} ",e.getMessage(),e);
			//发生异常了,要重新连接吧 30秒之后重试,如果失败就一直重试....
			ss.schedule(()->initMQ(), 30, TimeUnit.SECONDS);
			return;
		}
		if(consumer instanceof BasePlantformKafkaConsumer){
			BasePlantformKafkaConsumer bpkc = (BasePlantformKafkaConsumer) consumer;
			bpkc.setMessageConsuemr(new MQConsumer());
			consumer.startAsync(ss);
		}
		
	}
	@Override
	public void destroy() throws Exception {
		if(consumer!=null){
			try{
				consumer.stop();
			}catch(Exception e){
				logException(e);
			}
		}
	}
	private final class R implements LabeledRecord{
		final String key;
		final String label;
		final String content;
		private R(String key,String content,String label){
			this.label = label;
			this.key = key;
			this.content = content;
		}
		@Override
		public String getKey() {
			return key;
		}

		@Override
		public String getContent() {
			return content;
		}

		@Override
		public String getLabel() {
			return label;
		}
		public String toString(){
			return String.format("key:%s,label:%s,content:%s", key,label,content);
		}
		
	}
	BeanFactory bf;
	@Override
	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
		this.bf = beanFactory;
	}

}
